package com.cxf.cxfdaybat.config;

import com.cxf.cxfdaybat.constant.EncodingConstant;
import com.cxf.cxfdaybat.constant.FilePathConstant;
import com.cxf.cxfdaybat.constant.FileSeparatorConstant;
import com.cxf.cxfdaybat.entity.Person;
import com.cxf.cxfdaybat.utils.POJOConvertUtils;
import org.hibernate.validator.HibernateValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.MultiResourceItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.validator.ValidatingItemProcessor;
import org.springframework.batch.item.validator.ValidationException;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import javax.validation.ConstraintViolation;
import javax.validation.Validation;
import java.util.Set;

/**
 * description: PersonConfiguration <br>
 * date: 2021/11/12 19:07 <br>
 * author: cxf <br>
 * version: 1.0 <br>
 */
@Configuration
@EnableBatchProcessing // 开启批处理的支持
public class PersonConfiguration {


/*    @Value("D:\\tmp\\newFile\\person-20211113*.csv")
    private Resource[] inputResources;*/


    private Logger logger = LoggerFactory.getLogger(CsvBatchConfig.class);

    @Autowired
    private DataSource dataSource;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Bean
    public SimpleJobLauncher personJobLauncher() throws Exception{
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        // 设置jobRepository
        jobLauncher.setJobRepository(personJobRepository());

        // 异步运行
       //  jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor() );

        return jobLauncher;
    }


    @Bean
    public JobRepository personJobRepository() throws Exception{
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDataSource(dataSource);
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * @Title: personJob
     * @Description: 定义personJob
     * @Author: cxf
     * @DateTime: 2021/11/12 20:58
     * @param jobs
     * @return org.springframework.batch.core.Job
     * @throws
     */
    @Bean
    public Job personJob(JobBuilderFactory jobs){
        return jobs.get("personJob")
                .incrementer(new RunIdIncrementer())
                .flow(personStep())
                .end()
                .listener(personJobListener())
                .build();
    }

    /**
     * @Title: personJobListener
     * @Description: 定义Listener监听器
     * @Author: cxf
     * @DateTime: 2021/11/12 20:45
     * @param
     * @return org.springframework.batch.core.JobExecutionListener
     * @throws
     */
    @Bean
    public JobExecutionListener personJobListener(){
        return new JobExecutionListener(){
            private long startTime;
            private long endTime;

            @Override
            public void beforeJob(JobExecution jobExecution) {
                startTime = System.currentTimeMillis();
                logger.info("job process start...");
            }

            @Override
            public void afterJob(JobExecution jobExecution) {
                endTime = System.currentTimeMillis();
                logger.info("job process end...");
                logger.info("elapsed time: " + (endTime - startTime) + "ms");
            }
        };
    }

    /**
     * @Title: personStep
     * @Description: step定义：步骤包括ItemReader->ItemProcessor->ItemWriter 即读取数据->处理校验数据->写入数据
     * @Author: cxf
     * @DateTime: 2021/11/12 20:45
     * @return org.springframework.batch.core.Step
     * @throws
     */
    @Bean
    public Step personStep(){
        return stepBuilderFactory
                .get("personStep")
                .<Person, Person>chunk(65000) // Chunk的机制(即每次读取一条数据，再处理一条数据，累积到一定数量后再一次性交给writer进行写入操作)
                .reader(personMutiReader())
                .processor(personProcessor())
                .writer(personWriter())
                .build();

    }


    @Bean
    public ItemReader<Person> personMutiReader()   {


        // 使用FlatFileItemReader去读cvs文件，一行即一条数据
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        // 设置文件处在路径
        // reader.setResource(new ClassPathResource("person.csv"));
        reader.setResource(new FileSystemResource(FilePathConstant.PERSONPATH));
        // 设置编码格式
        reader.setEncoding(EncodingConstant.UTF_8);
        // entity与csv数据做映射
        reader.setLineMapper(new DefaultLineMapper<Person>() {
            {
                // FileSeparatorConstant.SEPARATOR_COMMA 设置文件的字段分隔符
                setLineTokenizer(new DelimitedLineTokenizer(FileSeparatorConstant.SEPARATOR_COMMA) {
                    {
                        //setNames(new String[]{"id", "name", "age", "gender"});
                        setNames(POJOConvertUtils.pojoToStrings(Person.class));
                    }
                });
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
                    {
                        setTargetType(Person.class);
                    }
                });
            }
        });

        MultiResourceItemReader itemReader = new MultiResourceItemReader();
        itemReader.setDelegate(reader);
        // 多路径下多文件
        FileSystemResource[] fileSystemResource = new FileSystemResource[]{
                new FileSystemResource("D:\\tmp\\newFile\\person1.csv"),
                new FileSystemResource("D:\\tmp\\person.csv")
        };
        // 相同路径下 文件同类型文件
/*        FileSystemResource[] fileSystemResource = new FileSystemResource[]{
                new FileSystemResource("D:\\tmp\\newFile\\person-20211113-*.csv")
        };*/

        itemReader.setResources(fileSystemResource);
        return itemReader;
    }




    /**
     * @Title: personReader
     * @Description: ItemReader定义：读取文件数据+entirty映射
     * @Author: cxf
     * @DateTime: 2021/11/12 20:47
     * @param
     * @return org.springframework.batch.item.ItemReader<com.cxf.cxfdaybat.entity.Person>
     * @throws
     */
    @Bean
    public ItemReader<Person> personSingleReader()   {

        // 使用FlatFileItemReader去读cvs文件，一行即一条数据
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        // 设置文件处在路径
        // reader.setResource(new ClassPathResource("person.csv"));
        reader.setResource(new FileSystemResource(FilePathConstant.PERSONPATH));
        // 设置编码格式
        reader.setEncoding(EncodingConstant.UTF_8);
        // 告诉 file reader 有多少标题行需要跳过。 通常CSV文件的第一行包含标题信息,如列名称,所以本例中让 reader 跳过文件的第一行
        reader.setLinesToSkip(1);
        // entity与csv数据做映射
        reader.setLineMapper(new DefaultLineMapper<Person>() {
            {
                // FileSeparatorConstant.SEPARATOR_COMMA 设置文件的字段分隔符
                // LineTokenizer 指定了如何将一行拆分为多个字段
                setLineTokenizer(new DelimitedLineTokenizer(FileSeparatorConstant.SEPARATOR_COMMA) {
                    {
                        //setNames(new String[]{"id", "name", "age", "gender"});
                        setNames(POJOConvertUtils.pojoToStrings(Person.class));
                    }
                });
                // fieldSetMapper 根据字段值构造一个对象
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
                    {
                        setTargetType(Person.class);
                    }
                });
            }
        });
        return reader;
    }

    /**
     * @Title: personProcessor
     * @Description: 处理数据+校验数据
     * @Author: cxf
     * @DateTime: 2021/11/12 20:47
     * @param
     * @return org.springframework.batch.item.ItemProcessor<com.cxf.cxfdaybat.entity.Person,com.cxf.cxfdaybat.entity.Person>
     * @throws
     */
    @Bean
    public ItemProcessor<Person, Person> personProcessor(){
        ValidatingItemProcessor<Person> validatingItemProcessor = new ValidatingItemProcessor<Person>() {

            @Override
            public Person process(Person item) throws ValidationException {
                // 执行super.process()才能调用自定义的校验器
                logger.info("processor start validating...");
                logger.info("now you can see me "+item.toString());
                super.process(item);
                // 数据处理，比如将中文性别设置为M/F
                if ("男".equals(item.getGender())) {
                    item.setGender("M");
                } else {
                    item.setGender("F");
                }
                logger.info("processor end validating...");
                return item;
            }
        };
        validatingItemProcessor.setFilter(false);
        validatingItemProcessor.setValidator(new Validator<Person>(){

            public  javax.validation.Validator validator = Validation.byProvider(HibernateValidator.class)
                                                            .configure()
                                                            .failFast(false)
                                                            .buildValidatorFactory()
                                                            .getValidator();

            @Override
            public void validate(Person person) throws ValidationException {

                Set<ConstraintViolation<Person>> validate = validator.validate(person);

                if (validate.size() > 0) {
                    StringBuilder message = new StringBuilder();
                    for (ConstraintViolation<Person> constraintViolation: validate) {
                        message.append(constraintViolation.getMessage() + "\n");
                    }
                    throw new ValidationException(message.toString());
                }
            }
        });
        return validatingItemProcessor;
    }

    /**
     * @Title: personWriter
     * @Description: ItemWriter定义：指定datasource，设置批量插入sql语句，写入数据库
     * @Author: cxf
     * @DateTime: 2021/11/12 20:47
     * @return org.springframework.batch.item.ItemWriter<com.cxf.cxfdaybat.entity.Person>
     * @throws
     */
    @Bean
    public ItemWriter<Person> personWriter(){
        // 使用jdbcBcatchItemWrite写数据到数据库中
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        // 设置有参数的sql语句
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
/*        String sql = "insert into person values(:id,:name,:age,:gender)";
        writer.setSql(sql);*/
        writer.setSql(POJOConvertUtils.pojotoSql(Person.class));
        writer.setDataSource(dataSource);

        return writer;
    }



}
